import pandas as pd
from manager.analysis import  *
from manager.conn import *
from manager.models import *
from manager.stores import *
from django.db.models import Count
from django.db.models import Sum
from django.db.models.functions import ExtractYear, ExtractMonth
from django.db.models import Case
from django.db.models import When
from django.db import connection
import json
import queue

#存储任务函数返回值
message_queue = queue.Queue()

#从数据源更新数据
# status：执行状态
# msg：执行信息
# id：任务id
# num：同步数据条数
def update_source(request,task_id):
    data = get_source_data(request)
    if data is None:
        msg = '数据源异常'
        id = task_id
        num = 0
        status = False
        data = {
            'msg':msg,
            'id':id,
            'num':num,
            'status':status
        }
        message_queue.put(data)

    else:
        type = request.POST.get('type')
        if type == 'wl':
            correct,error = handle_logistics_info(data)
            if correct is None and error is None:
                msg = '数据清洗异常'
                id = task_id
                num = 0
                status = False
                data = {
                    'msg':msg,
                    'id':id,
                    'num':num,
                    'status':status
                }
                message_queue.put(data)
            else:
                num = TradeInfo_store(correct)
                msg = '执行成功'
                id = task_id
                num = num
                status = False
                data = {
                    'msg':msg,
                    'id':id,
                    'num':num,
                    'status':status
                }
                message_queue.put(data)
        elif type == 'yh':
            correct,error = handle_userinfo(data)
            if correct is None and error is None:
                msg = '数据清洗异常'
                id = task_id
                num = 0
                status = False
                data = {
                    'msg':msg,
                    'id':id,
                    'num':num,
                    'status':status
                }
                message_queue.put(data)
            else:
                num = Client_store(correct)
                msg = '执行成功'
                id = task_id
                num = num
                status = False
                data = {
                    'msg':msg,
                    'id':id,
                    'num':num,
                    'status':status
                }
                message_queue.put(data)
        elif type == 'zh':
            correct,error = handle_other_info(data,'zh')
            if correct is None and error is None:
                msg = '数据清洗异常'
                id = task_id
                num = 0
                status = False
                data = {
                    'msg':msg,
                    'id':id,
                    'num':num,
                    'status':status
                }
                message_queue.put(data)
            else:
                num = Ship_store(correct)
                msg = '执行成功'
                id = task_id
                num = num
                status = False
                data = {
                    'msg':msg,
                    'id':id,
                    'num':num,
                    'status':status
                }
                message_queue.put(data)
        elif type == 'xh':
            correct,error = handle_other_info(data,'xh')
            if correct is None and error is None:
                msg = '数据清洗异常'
                id = task_id
                num = 0
                status = False
                data = {
                    'msg':msg,
                    'id':id,
                    'num':num,
                    'status':status
                }
                message_queue.put(data)
            else:
                num = Unship_store(correct)
                msg = '执行成功'
                id = task_id
                num = num
                status = False
                data = {
                    'msg':msg,
                    'id':id,
                    'num':num,
                    'status':status
                }
                message_queue.put(data)
        elif type == 'wlgs':
            correct,error = handle_other_info(data,'wlgs')
            if correct is None and error is None:
                msg = '数据清洗异常'
                id = task_id
                num = 0
                status = False
                data = {
                    'msg':msg,
                    'id':id,
                    'num':num,
                    'status':status
                }
                message_queue.put(data)
            else:
                num = TradeCompany_store(correct)
                msg = '执行成功'
                id = task_id
                num = num
                status = False
                data = {
                    'msg':msg,
                    'id':id,
                    'num':num,
                    'status':status
                }
                message_queue.put(data)
        elif type == 'jzx':
            correct,error = handle_other_info(data,'jzx')
            if correct is None and error is None:
                msg = '数据清洗异常'
                id = task_id
                num = 0
                status = False
                data = {
                    'msg':msg,
                    'id':id,
                    'num':num,
                    'status':status
                }
                message_queue.put(data)
            else:
                num = container_store(correct)
                msg = '执行成功'
                id = task_id
                num = num
                status = False
                data = {
                    'msg':msg,
                    'id':id,
                    'num':num,
                    'status':status
                }
                message_queue.put(data)
    
# 处理用户信息
def handle_userinfo(data,column_name = '客户编号'):
    # 判断数据格式
    columns = list(data.columns)
    col_list = ('客户名称','客户编号','手机号','省市区')
    columns = data.columns
    if len(columns) < len(col_list):
        return None,None
    if set(col_list).issubset(data.columns) is False:
        return None,None

    # # 信息去重
    drop_duplicate(data=data,column_name='客户编号')

    # 去除空值行
    error_data_null = remove_rows_with_null_values(data)

    # 身份证检验
    correct_data,error_data_id = check_id(data=data,column_name='客户编号')

    error_data = pd.concat([error_data_id,error_data_null])
    return correct_data,error_data

# 处理物流信息
def handle_logistics_info(logist_data,user_id):
    # 判断数据格式
    # columns = list(logist_data.columns)
    col_list = ('提单号','货主名称','货主代码','物流公司（货代）','集装箱箱号','货物名称','货重（吨）')
    columns = logist_data.columns
    if len(columns) < len(col_list):
        return None,None
    if set(col_list).issubset(logist_data.columns) is False:
        return None,None
    # 去除空值行
    error_data_null = remove_rows_with_null_values(logist_data)

    # 检验物流信息中客户是否在客户列表中
    # 查询数据库中用户信息
    query_data = Client.objects.filter(user=user_id)

    # 将查询结果转换为字典列表
    data_dict_list = list(query_data.values())

    # 将字典列表转换为DataFrame
    client_data = pd.DataFrame(data_dict_list)

    # 得到正确和错误的物流信息
    correct_data,error_data_id = judge_common_user(client_data,logist_data)

    # 合并错误数据
    error_data = pd.concat([error_data_id, error_data_null])

    return correct_data,error_data

# 处理其他信息
def handle_other_info(other_data,type_data):   
    # 去除空值行
    error_data_null = remove_rows_with_null_values(other_data)
    if type_data == 'jzx':
        # 判断数据格式
        col_list = ('堆存港口','集装箱箱号', '箱尺寸（TEU）','提单号', '堆场位置', '操作','操作日期')
        columns = other_data.columns
        if len(columns) < len(col_list):
            return None,None
        if set(col_list).issubset(other_data.columns) is False:
            return None,None
        
        try:
            other_data['操作日期'] = pd.to_datetime(other_data['操作日期'])
            other_data['箱尺寸（TEU）'] = other_data['箱尺寸（TEU）'].astype(int)
        except Exception:
            a = 1

    elif type_data == 'zh' or type_data =='xh':
        # 判断数据格式
        col_list = ('船公司','船名称','作业开始时间','作业结束时间','始发时间','到达时间', '作业港口', '提单号', '集装箱箱号', '箱尺寸（TEU）','启运地', '目的地')
        columns = other_data.columns
        if len(columns) < len(col_list):
            return None,None
        if set(col_list).issubset(other_data.columns) is False:
            return None,None
        try:
            # 提取日期部分并转换为日期格式
            # other_data['作业开始时间'] = pd.to_datetime(other_data['作业开始时间'].str[:10])
            # other_data['作业结束时间'] = pd.to_datetime(other_data['作业结束时间'].str[:10])
            # other_data['始发时间'] = pd.to_datetime(other_data['始发时间'].str[:10])
            # other_data['到达时间'] = pd.to_datetime(other_data['到达时间'].str[:10])
            other_data['作业开始时间'] = other_data['作业开始时间'].str[:10]
            other_data['作业结束时间'] = other_data['作业结束时间'].str[:10]
            other_data['始发时间'] = other_data['始发时间'].str[:10]
            other_data['到达时间'] = other_data['到达时间'].str[:10]
        except Exception:
            a = 1  
            
    elif type_data == 'wlgs':
        col_list = ('公司名称','客户编号','联系人','电话','省市区')
        columns = other_data.columns
        if len(columns) < len(col_list):
            return None,None
        if set(col_list).issubset(other_data.columns) is False:
            return None,None
    return  other_data,error_data_null
    
# 数据聚合
def handle_group_info(request,user_id):
    # 列名
    column_name = request.POST.get('column_name')
    table = request.POST.get('table')
    # 将字段名映射成对应的数据库字段名
    if table == "wl":
        field_mapping = {
            '提单号': 'tradeid',
            '货主名称': 'client',
            '货主代码': 'clientid',
            '物流公司（货代）': 'company',
            '集装箱箱号': 'container',
            '货物名称': 'goods',
            '货重（吨）': 'weight',
        }
        target_column_name = field_mapping.get(column_name)
        try:
            queryset = TradeInfo.objects.filter(user=user_id).values(target_column_name).annotate(count=Count(target_column_name))
            # 将查询结果转换为 DataFrame
            df = pd.DataFrame.from_records(queryset)
            # 重命名 DataFrame 的列名
            if df is not None:
                df.rename(columns={target_column_name: column_name, 'count': '数量'}, inplace=True) 
            return df
        except Exception as err:
            print('util.py line 150',str(err))
            return None
    elif table == 'zh':
        field_mapping = {
            '船公司': 'company',
            '船名称': 'shipname',
            '作业开始时间': 'btime',
            '作业结束时间': 'etime',
            '始发时间': 'gotime',
            '到达时间': 'artime',
            '作业港口': 'harbor',
            '提单号': 'sid',
            '集装箱箱号': 'con_id',
            '箱尺寸（TEU）': 'size',
            '启运地': 'source',
            '目的地': 'destination',
        }
        target_column_name = field_mapping.get(column_name)
        try:
            queryset = Ship.objects.filter(user=user_id).values(target_column_name).annotate(count=Count(target_column_name))
            # 将查询结果转换为 DataFrame
            df = pd.DataFrame.from_records(queryset)
            # 重命名 DataFrame 的列名
            if df is not None:
                df.rename(columns={target_column_name: column_name, 'count': '数量'}, inplace=True) 
            return df
        except Exception as err:
            return None
    elif table == 'xh':
        field_mapping = {
            '船公司': 'company',
            '船名称': 'shipname',
            '作业开始时间': 'btime',
            '作业结束时间': 'etime',
            '始发时间': 'gotime',
            '到达时间': 'artime',
            '作业港口': 'harbor',
            '提单号': 'sid',
            '集装箱箱号': 'con_id',
            '箱尺寸（TEU）': 'size',
            '启运地': 'source',
            '目的地': 'destination',
        }
        target_column_name = field_mapping.get(column_name)
        try:
            queryset = Unship.objects.filter(user=user_id).values(target_column_name).annotate(count=Count(target_column_name))
            # 将查询结果转换为 DataFrame
            df = pd.DataFrame.from_records(queryset)
            # 重命名 DataFrame 的列名
            if df is not None:
                df.rename(columns={target_column_name: column_name, 'count': '数量'}, inplace=True) 
            return df
        except Exception as err:
            return None
    elif table == 'jzx':
        field_mapping = {
            '堆存港口': 'harbor',
            '集装箱箱号': 'con_id',
            '箱尺寸（TEU）': 'size',
            '提单号': 'sid',
            '堆场位置': 'store',
            '操作': 'op',
            '操作日期': 'opdata',
        }
        target_column_name = field_mapping.get(column_name)
        try:
            queryset = Container.objects.filter(user=user_id).values(target_column_name).annotate(count=Count(target_column_name))
            # 将查询结果转换为 DataFrame
            df = pd.DataFrame.from_records(queryset)
            # 重命名 DataFrame 的列名
            if df is not None:
                df.rename(columns={target_column_name: column_name, 'count': '数量'}, inplace=True) 
            return df
        except Exception as err:
            print('util.py handle_group',str(err))
            return None


# 从达梦数据中返回测试数据:
def get_store_data(request,user_id):
    table = request.POST.get('table')
    try:
        if table == 'wl':
            queryset = TradeInfo.objects.filter(user=user_id)[:10]
            df = pd.DataFrame.from_records(queryset.values())
            if df is not None:
                df.rename(columns={'tradeid': '提单号','client':'货主名称','clientid':'货主代码','company':'物流公司（货代）','container':'集装箱箱号','goods':'货物名称','weight':'货重（吨）'}, inplace=True)
            return df
        elif table =='zh':
            queryset = Ship.objects.filter(user=user_id)[:10]  
            df = pd.DataFrame.from_records(queryset.values())
            if df is not None:
                df['btime'] = pd.to_datetime(df['btime'])  # 将时间戳转换为日期时间类型
                df['etime'] = pd.to_datetime(df['etime'])  # 将时间戳转换为日期时间类型
                df['artime'] = pd.to_datetime(df['artime'])  # 将时间戳转换为日期时间类型
                df['gotime'] = pd.to_datetime(df['gotime'])  # 将时间戳转换为日期时间类型
                #将日期时间类型字段格式化为指定的日期字符串（例如 'YYYY-MM-DD'）
                df['btime'] = df['btime'].dt.strftime('%Y-%m-%d')
                df['etime'] = df['etime'].dt.strftime('%Y-%m-%d')
                df['gotime'] = df['gotime'].dt.strftime('%Y-%m-%d')
                df['artime'] = df['artime'].dt.strftime('%Y-%m-%d')

                df.rename(columns={'company': '船公司','shipname':'船名称', 'btime': '作业开始时间','etime': '作业结束时间', 'gotime':'始发时间','artime':'到达时间','harbor':'作业港口','sid':'提单号','con_id':'集装箱箱号',
            'size': '箱尺寸（TEU）', 'source':'启运地','destination':'目的地'}, inplace=True)
            return df
        elif table =='xh':
            queryset = Unship.objects.filter(user=user_id)[:10]
            df = pd.DataFrame.from_records(queryset.values())
            if df is not None:
                df['btime'] = pd.to_datetime(df['btime'])  # 将时间戳转换为日期时间类型
                df['etime'] = pd.to_datetime(df['etime'])  # 将时间戳转换为日期时间类型
                df['artime'] = pd.to_datetime(df['artime'])  # 将时间戳转换为日期时间类型
                df['gotime'] = pd.to_datetime(df['gotime'])  # 将时间戳转换为日期时间类型
                #将日期时间类型字段格式化为指定的日期字符串（例如 'YYYY-MM-DD'）
                df['btime'] = df['btime'].dt.strftime('%Y-%m-%d')
                df['etime'] = df['etime'].dt.strftime('%Y-%m-%d')
                df['gotime'] = df['gotime'].dt.strftime('%Y-%m-%d')
                df['artime'] = df['artime'].dt.strftime('%Y-%m-%d')

                df.rename(columns={'company': '船公司','shipname':'船名称', 'btime': '作业开始时间','etime': '作业结束时间', 'gotime':'始发时间','artime':'到达时间','harbor':'作业港口','sid':'提单号','con_id':'集装箱箱号',
            'size': '箱尺寸（TEU）', 'source':'启运地','destination':'目的地'}, inplace=True) 
            return df
        elif table =='jzx':
            queryset = Container.objects.filter(user=user_id)[:10]
            df = pd.DataFrame.from_records(queryset.values()) 
            if df is not None:
                df['opdata'] = pd.to_datetime(df['opdata'])  # 将时间戳转换为日期时间类型
                #将日期时间类型字段格式化为指定的日期字符串（例如 'YYYY-MM-DD'）
                df['opdata'] = df['opdata'].dt.strftime('%Y-%m-%d')
                df.rename(columns={'harbor':'堆存港口','con_id':'集装箱箱号','size':'箱尺寸（TEU）','sid':'提单号','store':'堆场位置','op':'操作','opdata':'操作日期' }, inplace=True) 
            return df
        elif table =='wlgs':
            queryset = TradeCompany.objects.filter(user=user_id)[:10]  
            df = pd.DataFrame.from_records(queryset.values())
            if df is not None:
                df.rename(columns={'company': '公司名称','clientid':'客户编号', 'admin': '联系人','tel': '电话', 'area':'省市区'}, inplace=True)
            return df
        elif table =='yh':
            queryset = Client.objects.filter(user=user_id)[:10]  
            df = pd.DataFrame.from_records(queryset.values())
            if df is not None:
                df.rename(columns={'name': '客户名称','clientid':'客户编号', 'tel': '手机号','addr': '省市区'}, inplace=True)
            return df
    except Exception as err:
        print(err)
        return None
# 从数据源读取数据，以dataframe形式返回
def get_source_data(request,user_id=''):
    # 判断数据来源
    src_id = request.POST.get('id')
    src_obj = DataSrc.objects.filter(id = src_id).first()
    source_type = src_obj.type
    # mysql数据源
    if source_type == 'mysql':
        # 获取数据源参数
        src_data = json.loads(src_obj.json)
        host = src_data['host']
        port = int(src_data['port'])
        username = src_data['username']
        password = src_data['password']
        database = src_data['database']
        table = request.POST.get('table')
       # 采集数据
        try:
            data = col_mysql(host=host, port=port, username=username, password=password, database=database, table=table,
                             getdata=True)
            if data is None:
                return None
            else:
                return data
        except Exception as err:
            print('mysql连接异常',str(err))
            return None
    # minio数据源
    elif source_type == 'minio':
        # 获取数据源参数
        src_data = json.loads(src_obj.json)
        host = src_data['host']
        port = str(src_data['port'])
        access_key = src_data['acc']
        secret_key = src_data['sec']
        bucket = src_data['bucket']
        ssl = src_data['ssl']
        object_name = request.POST.get('object')
        try:
            # 采集数据
            data = col_minio(minio_host=host, minio_port=port, access_key=access_key, secret_key=secret_key,
                             bucket=bucket, object_name=object_name, use_ssl=ssl)
            if data is None:
                return None
            else:
                return data
        except Exception as err:
            print('minio连接异常',str(err))
            return None
    # hdfs数据源
    elif source_type == 'hdfs':
        # 获取数据源参数
        src_data = json.loads(src_obj.json)
        host = src_data['host']
        port = int(src_data['port'])
        user = src_data['user']
        path = request.POST.get('path')

        try:
            # 采集数据
            data = col_hdfs(hdfs_host=host, hdfs_port=port, user=user, hdfs_path=path)
            if data is None:
                return None
            else:
                return data
        except Exception as err:
            print('hdfs连接异常',str(err))
            return None

    elif source_type == 'file':
        file_id = request.POST.get('file_id')
        user_directory = f'storefile//{user_id}'
        file_path = user_directory + '//' + file_id
        with open(file_path, 'rb') as file:
            encodings_to_try = [
            'utf-8', 'gb18030', 'gbk', 'big5', 'shift_jis', 'euc-jp', 'euc-kr', 'iso-8859-1', 'windows-1252']
            # 检查文件格式
            allowed_formats = ['csv', 'txt', 'xlsx']
            file_type = file.name.split('.')[-1]
            if file_type not in allowed_formats:
                return None
            for encoding_to_try in encodings_to_try:
                try:
                    file.seek(0)
                    if file_type == "txt":
                        # 处理txt文件
                        df = pd.read_csv(file,delimiter='\t', encoding=encoding_to_try)
                        df_new = df[df.columns[0]].str.split(r"',|\",", expand=True)
                        #把列名里的\t和"和'去掉
                        df.rename(columns=lambda x: re.sub(r'[\"\t\']', '',x), inplace=True)
                        #把列名按照,切割成多列
                        df_new.columns =df.columns[0].split(',')
                        #去掉所有的'和"
                        df_new=df_new.replace(['\'', '\"'], '', regex=True)
                        return df_new
                        return dataframe
                    elif file_type == "csv":
                        # 处理csv文件
                        df = pd.read_csv(file, encoding=encoding_to_try)
                        df=df.replace(['\'', '\"'], '', regex=True)
                        return df
                    elif file_type == 'xlsx':
                        df = pd.read_excel(file)
                        return df

                except UnicodeDecodeError:
                    continue
            return None

#异常数据处理
def handle_out(request,user_id):
    table = request.POST.get('table')
    column_name = request.POST.get('column_name')
    upper = int(request.POST.get('upper'))
    lower = int(request.POST.get('lower'))
    # 将字段名映射成对应的数据库字段名
    if table == "wl":
        if column_name != '货重（吨）':
            return False
        try:
            query = """
            SELECT *
            FROM tradeinfo
            WHERE 
            (weight > %s or weight < %s)
            AND user_id = %s
            LIMIT 
            20
            """

            with connection.cursor() as cursor:
                cursor.execute(query, [upper,lower,user_id])
                results = cursor.fetchall()
            df = pd.DataFrame.from_records(results)
            df.columns = ['id','提单号','货主名称','货主代码','物流公司（货代）','集装箱箱号','货物名称','货重（吨）','user_id']
            return df
        except Exception as err:
            print('util.py def handle_out',str(err))
            return None
    elif table == 'zh':
        if column_name != '箱尺寸（TEU）':
            return False
        try:
            query = """
            SELECT *
            FROM ship
            WHERE 
            (size > %s or size < %s)
            AND user_id = %s
            LIMIT
            20
            """

            with connection.cursor() as cursor:
                cursor.execute(query, [upper,lower,user_id])
                results = cursor.fetchall()
            df = pd.DataFrame.from_records(results)
            # 重命名 DataFrame 的列名
            df.columns = ['id','船公司','船名称','作业开始时间','作业结束时间','始发时间','到达时间','作业港口','提单号','集装箱箱号','箱尺寸（TEU）','启运地','目的地','user_id']
        
            df['作业开始时间'] = pd.to_datetime(df['作业开始时间'])  # 将时间戳转换为日期时间类型
            df['作业结束时间'] = pd.to_datetime(df['作业结束时间'])  # 将时间戳转换为日期时间类型
            df['到达时间'] = pd.to_datetime(df['到达时间'])  # 将时间戳转换为日期时间类型
            df['始发时间'] = pd.to_datetime(df['始发时间'])  # 将时间戳转换为日期时间类型
            #将日期时间类型字段格式化为指定的日期字符串（例如 'YYYY-MM-DD'）
            df['作业开始时间'] = df['作业开始时间'].dt.strftime('%Y-%m-%d')
            df['作业结束时间'] = df['作业结束时间'].dt.strftime('%Y-%m-%d')
            df['始发时间'] = df['始发时间'].dt.strftime('%Y-%m-%d')
            df['到达时间'] = df['到达时间'].dt.strftime('%Y-%m-%d')
        
        except Exception as err:
                print(str(err))
                return None
    elif table == 'xh':
        if column_name != '箱尺寸（TEU）':
            return False
        try:
            query = """
            SELECT *
            FROM unship
            WHERE 
            (size > %s or size < %s)
            AND user_id = %s
            LIMIT
            20
            """

            with connection.cursor() as cursor:
                cursor.execute(query, [upper,lower,user_id])
                results = cursor.fetchall()
            df = pd.DataFrame.from_records(results)
            # 重命名 DataFrame 的列名
            df.columns = ['id','船公司','船名称','作业开始时间','作业结束时间','始发时间','到达时间','作业港口','提单号','集装箱箱号','箱尺寸（TEU）','启运地','目的地','user_id']
        
            df['作业开始时间'] = pd.to_datetime(df['作业开始时间'])  # 将时间戳转换为日期时间类型
            df['作业结束时间'] = pd.to_datetime(df['作业结束时间'])  # 将时间戳转换为日期时间类型
            df['到达时间'] = pd.to_datetime(df['到达时间'])  # 将时间戳转换为日期时间类型
            df['始发时间'] = pd.to_datetime(df['始发时间'])  # 将时间戳转换为日期时间类型
            #将日期时间类型字段格式化为指定的日期字符串（例如 'YYYY-MM-DD'）
            df['作业开始时间'] = df['作业开始时间'].dt.strftime('%Y-%m-%d')
            df['作业结束时间'] = df['作业结束时间'].dt.strftime('%Y-%m-%d')
            df['始发时间'] = df['始发时间'].dt.strftime('%Y-%m-%d')
            df['到达时间'] = df['到达时间'].dt.strftime('%Y-%m-%d')
            return df
        except Exception as err:
            return None
    elif table == 'jzx':
        if column_name != '箱尺寸（TEU）':
            return False
        try:
            query = """
            SELECT *
            FROM container
            WHERE 
            (size > %s or size < %s)
            AND user_id = %s
            LIMIT
            20
            """

            with connection.cursor() as cursor:
                cursor.execute(query, [upper,lower,user_id])
                results = cursor.fetchall()
            df = pd.DataFrame.from_records(results)
            # 重命名 DataFrame 的列名
            df.columns = ['id','堆存港口','集装箱箱号','箱尺寸（TEU）','提单号','堆场位置','操作','操作日期','user_id']
            df['操作日期'] = pd.to_datetime(df['操作日期'])  # 将时间戳转换为日期时间类型
            #将日期时间类型字段格式化为指定的日期字符串（例如 'YYYY-MM-DD'）
            df['操作日期'] = df['操作日期'].dt.strftime('%Y-%m-%d')
            return df
        except Exception as err:
            print('util.py line 241',str(err))
            return None

# 港口吞吐量
def port_throughput(request,user_id):
    port = request.POST.get('port')
    year = int(request.POST.get('year'))
    query = """
        SELECT
            EXTRACT(MONTH FROM opdata) AS 月,
            SUM(CASE WHEN op = '入库' THEN size ELSE 0 END) AS 入库,
            SUM(CASE WHEN op = '出库' THEN size ELSE 0 END) AS 出库
        FROM
            container
        WHERE 
            user_id = %s
            AND harbor = %s
            AND EXTRACT(YEAR FROM opdata) = %s
        GROUP BY
            harbor,
            EXTRACT(YEAR FROM opdata),
            EXTRACT(MONTH FROM opdata)
        ORDER BY 
            EXTRACT(MONTH FROM opdata)
    """
    try:
        with connection.cursor() as cursor:
            cursor.execute(query, [user_id,port,year])
            results = cursor.fetchall()
            df = pd.DataFrame.from_records(results)
            if df is not None:
                df.columns = ['month','inputTEU','outputTEU']
            return df
    except Exception as err:
        print('util.py',str(err))
        return None

# 港口不同货物吞吐量占比
def port_goods_throughput(request,user_id):
    try:
        port = request.POST.get('port')
        # 入库不同货物的总重量
        inbound_weights = TradeInfo.objects.filter(user = user_id ,container__in=Container.objects.filter(op='入库', harbor=port, user=user_id).values('con_id')).values('goods').annotate(total_weight=Sum('weight'))
        # 出库不同货物的总重量
        outbound_weights = TradeInfo.objects.filter(user = user_id, container__in=Container.objects.filter(op='出库', harbor=port, user=user_id).values('con_id')).values('goods').annotate(total_weight=Sum('weight'))
        # 入库所有货物的总重量
        in_weight_sum = sum(item['total_weight'] for item in inbound_weights)
        # 出库所有货物的总重量
        out_weight_sum = sum(item['total_weight'] for item in outbound_weights)
        in_dic = []
        out_dic = []
        for item in inbound_weights:
            new_item = {}
            new_item['goods'] = item.get('goods')
            new_item['in_per'] = item.get('total_weight') / in_weight_sum *100
            in_dic.append(new_item)
        in_df = pd.DataFrame(in_dic)
        for item in outbound_weights:
            new_item = {}
            new_item['goods'] = item.get('goods')
            new_item['out_per'] = item.get('total_weight') / out_weight_sum *100
            out_dic.append(new_item)
        out_df = pd.DataFrame(out_dic)
    except Exception as err:
        print('util.py',str(err))
        return None,None
    return in_df,out_df

# 不同货物流向
def flow_analysis(request,user_id):
    sql = """
        SELECT
            tradeinfo.goods AS 货物名称,
            ship.source AS 启运地,
            ship.destination AS 目的地,
            SUM(ship.size) AS 总箱尺寸
        FROM
            tradeinfo, ship
        WHERE
            tradeinfo.user_id = %s
            AND tradeinfo.goods = %s
            AND ship.user_id = %s
            AND tradeinfo.tradeid = ship.sid AND tradeinfo.container = ship.con_id
        GROUP BY
            tradeinfo.goods, ship.source, ship.destination
        ORDER BY
            tradeinfo.goods, ship.source, ship.destination
    """
    goods = request.POST.get('goods')
    try:
        with connection.cursor() as cursor:
            cursor.execute(sql,[user_id,goods,user_id])
            results = cursor.fetchall()
            df = pd.DataFrame.from_records(results)
            if df is not None:
                df.columns=['goods', 'source','destination','number']
                result = df[df['goods'] == goods]
            return result
    except Exception as err:
        print('util.py def flow',str(err))
        return None

def process_value(value):
    if value > 1000:
        return np.nan  # 将大于1000的值置为null
    else:
        return value  

#港口出库入库环比
def round_analysis(request,user_id):
    harbor = request.POST.get('port')
    year = int(request.POST.get('year'))
    query = """
        SELECT
            t2.月,
            (t2.入库 - t2.同年上月入库) * 100 / (t2.同年上月入库 + 0.1)  AS 入库环比,
            (t2.出库 - t2.同年上月出库) * 100 / (t2.同年上月出库 + 0.1)  AS 出库环比,
            (t2.入库 - t2.上年同月入库) * 100 / (t2.上年同月入库 + 0.1)  AS 入库同比,
            (t2.出库 - t2.上年同月出库) * 100 / (t2.上年同月出库 + 0.1)  AS 出库同比
        FROM
            (SELECT 
                t1.年,
                t1.月,
                t1.入库,
                t1.出库,
                LAG(t1.入库, 1, 0) OVER(ORDER BY t1.年, t1.月) AS 同年上月入库,
                LAG(t1.出库, 1, 0) OVER(ORDER BY t1.年, t1.月) AS 同年上月出库,
                LAG(t1.入库, 12, 0) OVER(ORDER BY t1.年, t1.月) AS 上年同月入库, 
                LAG(t1.出库, 12, 0) OVER(ORDER BY t1.年, t1.月) AS 上年同月出库
            FROM
                (SELECT
                    EXTRACT(YEAR FROM opdata) AS 年,
                    EXTRACT(MONTH FROM opdata) AS 月,
                    SUM(CASE WHEN op = '入库' THEN 1 ELSE 0 END) AS 入库,
                    SUM(CASE WHEN op = '出库' THEN 1 ELSE 0 END) AS 出库
                FROM container
                WHERE 
                    harbor = %s
                    AND user_id = %s
                GROUP BY harbor, EXTRACT(YEAR FROM opdata), EXTRACT(MONTH FROM opdata)
                ORDER BY EXTRACT(YEAR FROM opdata), EXTRACT(MONTH FROM opdata)
            ) AS t1(年, 月, 入库, 出库)
        ) AS t2(年, 月, 入库, 出库, 同年上月入库, 同年上月出库, 上年同月入库, 上年同月出库)
        WHERE t2.年 = %s
    """
    try:
        # 执行参数化查询
        with connection.cursor() as cursor:
            cursor.execute(query, [harbor, user_id,year])
            rows = cursor.fetchall()
            df = pd.DataFrame.from_records(rows)
            if df is not None:
                df = df.iloc[:, -5:]
                df.columns =['month','in_huan','out_huan','in_tong','out_tong']
                # 对数据框的特定列应用条件判断函数
                df['in_huan'] = df['in_huan'].apply(lambda x: process_value(x))
                df['out_huan'] = df['out_huan'].apply(lambda x: process_value(x))
                df['in_tong'] = df['in_tong'].apply(lambda x: process_value(x))
                df['out_tong'] = df['out_tong'].apply(lambda x: process_value(x))
        return df
    except Exception as err:
        print('util.py def round_analysis',str(err))
        return None
    
# 港口货物的吞吐趋势
def port_goods_tendency(request,user_id):
    port = request.POST.get('port')
    year = int(request.POST.get('year'))
    goods = request.POST.get('goods')
    query = """
        SELECT
            MONTH(c.opdata) AS month,
            SUM(CASE WHEN c.op = '入库' THEN t.weight ELSE 0 END) AS 入库总重量,
            SUM(CASE WHEN c.op = '出库' THEN t.weight ELSE 0 END) AS 出库总重量
        FROM
            container c
        JOIN
            tradeinfo t ON c.sid = t.tradeid
        WHERE
            c.harbor = %s
            AND c.user_id = %s
            AND YEAR(c.opdata) = %s
            AND t.goods = %s
        GROUP BY
            MONTH(c.opdata)
        ORDER BY
            MONTH(c.opdata);
    """
    try:
        with connection.cursor() as cursor:
            cursor.execute(query, [port, user_id, year, goods])
            results = cursor.fetchall()
            df = pd.DataFrame.from_records(results)
            if df is not None:
                df.columns = ['month','in_total','out_total']
        return df
    except Exception as err:
        print(str(err))
        return None

# 流转周期
def cycle(request,user_id):
    query = """
        SELECT
        tradeinfo.goods AS 货物名称,
        AVG(t2.流转时间) AS 堆场流转周期
        FROM
        (SELECT *
        FROM
        (SELECT
        con_id,
        sid,
        op,
        opdata,
        LEAD(opdata,1,opdata + 1) OVER(ORDER BY con_id,opdata,op DESC,sid) - opdata AS 流转时间 
        FROM
        container) AS t1
        WHERE t1.op = '入库' AND 流转时间 >= 0 AND 流转时间 <= 3) AS t2,tradeinfo
        WHERE tradeinfo.tradeid = t2.sid
        GROUP BY tradeinfo.goods
    """
    try:
        with connection.cursor() as cursor:
            cursor.execute(query)
            results = cursor.fetchall()
            df = pd.DataFrame.from_records(results)
            if df is not None:
                df.columns = ['goods','times']
                df['times'] = df['times'].apply(lambda x: float(x) if x is not None else 0)
        return df
    except Exception as err:
        print(str(err))
        return None
# 计算物流景气指数
def cal_prosperity(request,user_id):
    data = port_throughput(request,user_id)
    total_inbound = data['inputTEU'].sum()
    total_outbound = data['outputTEU'].sum()
    data['mrate'] = (data['inputTEU'] + data['outputTEU']) / (total_inbound + total_outbound)
    rate_df = data[['month', 'mrate']]
    round_data = round_analysis(request,user_id)
    round_data.fillna(0, inplace=True)  # 将NaN替换为0
    round_data['in_huan'] = round_data['in_huan'].apply(lambda x: float(x) if x is not None else 0)
    round_data['out_huan'] = round_data['out_huan'].apply(lambda x: float(x) if x is not None else 0)
    round_data['in_tong'] = round_data['in_tong'].apply(lambda x: float(x) if x is not None else 0)
    round_data['out_tong'] = round_data['out_tong'].apply(lambda x: float(x) if x is not None else 0)
    round_data['sum'] = round_data['in_huan'] + round_data['out_huan'] + round_data['in_tong'] + round_data['out_tong']
    round_data['rate'] = round_data['sum'] * rate_df['mrate']
    result_df = round_data[['month', 'rate']]
    return result_df



def export_data(request):
    table = request.POST.get('table')
    user_id = request.POST.get('user_id')
    try:
        if table == 'wl':
            queryset = TradeInfo.objects.filter(user=user_id)[:500]
            df = pd.DataFrame.from_records(queryset.values())
            if df is not None:
                df.rename(columns={'tradeid': '提单号','client':'货主名称','clientid':'货主代码','company':'物流公司（货代）','container':'集装箱箱号','goods':'货物名称','weight':'货重（吨）'}, inplace=True)
            return df
        elif table =='zh':
            queryset = Ship.objects.filter(user=user_id)[:500]  
            df = pd.DataFrame.from_records(queryset.values())
            if df is not None:
                df['btime'] = pd.to_datetime(df['btime'])  # 将时间戳转换为日期时间类型
                df['etime'] = pd.to_datetime(df['etime'])  # 将时间戳转换为日期时间类型
                df['artime'] = pd.to_datetime(df['artime'])  # 将时间戳转换为日期时间类型
                df['gotime'] = pd.to_datetime(df['gotime'])  # 将时间戳转换为日期时间类型
                #将日期时间类型字段格式化为指定的日期字符串（例如 'YYYY-MM-DD'）
                df['btime'] = df['btime'].dt.strftime('%Y-%m-%d')
                df['etime'] = df['etime'].dt.strftime('%Y-%m-%d')
                df['gotime'] = df['gotime'].dt.strftime('%Y-%m-%d')
                df['artime'] = df['artime'].dt.strftime('%Y-%m-%d')

                df.rename(columns={'company': '船公司','shipname':'船名称', 'btime': '作业开始时间','etime': '作业结束时间', 'gotime':'始发时间','artime':'到达时间','harbor':'作业港口','sid':'提单号','con_id':'集装箱箱号',
            'size': '箱尺寸（TEU）', 'source':'启运地','destination':'目的地'}, inplace=True)
            return df
        elif table =='xh':
            queryset = Unship.objects.filter(user=user_id)[:500]
            df = pd.DataFrame.from_records(queryset.values())
            if df is not None:
                df['btime'] = pd.to_datetime(df['btime'])  # 将时间戳转换为日期时间类型
                df['etime'] = pd.to_datetime(df['etime'])  # 将时间戳转换为日期时间类型
                df['artime'] = pd.to_datetime(df['artime'])  # 将时间戳转换为日期时间类型
                df['gotime'] = pd.to_datetime(df['gotime'])  # 将时间戳转换为日期时间类型
                #将日期时间类型字段格式化为指定的日期字符串（例如 'YYYY-MM-DD'）
                df['btime'] = df['btime'].dt.strftime('%Y-%m-%d')
                df['etime'] = df['etime'].dt.strftime('%Y-%m-%d')
                df['gotime'] = df['gotime'].dt.strftime('%Y-%m-%d')
                df['artime'] = df['artime'].dt.strftime('%Y-%m-%d')

                df.rename(columns={'company': '船公司','shipname':'船名称', 'btime': '作业开始时间','etime': '作业结束时间', 'gotime':'始发时间','artime':'到达时间','harbor':'作业港口','sid':'提单号','con_id':'集装箱箱号',
            'size': '箱尺寸（TEU）', 'source':'启运地','destination':'目的地'}, inplace=True) 
            return df
        elif table =='jzx':
            queryset = Container.objects.filter(user=user_id)[:500]
            df = pd.DataFrame.from_records(queryset.values()) 
            if df is not None:
                df['opdata'] = pd.to_datetime(df['opdata'])  # 将时间戳转换为日期时间类型
                #将日期时间类型字段格式化为指定的日期字符串（例如 'YYYY-MM-DD'）
                df['opdata'] = df['opdata'].dt.strftime('%Y-%m-%d')
                df.rename(columns={'harbor':'堆存港口','con_id':'集装箱箱号','size':'箱尺寸（TEU）','sid':'提单号','store':'堆场位置','op':'操作','opdata':'操作日期' }, inplace=True) 
            return df
    except Exception as err:
        print(err)
        return None